package com.digitalchina.dcn.dcss.portal.server;

import com.digitalchina.dcn.dcss.portal.client.AuthInfo;
import com.digitalchina.dcn.dcss.portal.client.PortalClient;
import com.digitalchina.dcn.dcss.portal.protocol.PortalAttribute;
import com.digitalchina.dcn.dcss.portal.protocol.PortalMessage;
import com.digitalchina.dcn.dcss.portal.protocol.PortalMessage.AuthType;
import com.digitalchina.dcn.dcss.portal.protocol.PortalMessage.ErrorCode;
import com.digitalchina.dcn.dcss.portal.protocol.PortalMessage.Type;
import com.digitalchina.dcn.dcss.portal.protocol.PortalMessage.UserIP;
import com.digitalchina.dcn.dcss.portal.protocol.PortalMessageFactory;
import com.digitalchina.dcn.dcss.portal.util.PortalContext;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.DatagramChannel;
import net.jradius.util.CHAP;
import org.apache.spark.SparkConf;
import org.apache.spark.network.util.*;
import org.apache.spark.util.ThreadUtils;
import org.eclipse.jetty.util.MultiException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.context.request.async.DeferredResult;
import org.springframework.web.servlet.ModelAndView;

import java.io.Closeable;
import java.io.IOException;
import java.net.BindException;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created by Administrator on 2016/8/21.
 */
public class PortalServer implements Closeable, PortalMessageHandler, MessageSendCallback{

    private final Logger logger = LoggerFactory.getLogger(PortalServer.class);

    private final PortalContext context;
    private final SparkConf sparkConf = new SparkConf();
    private final TransportConf conf;
    private PortalClient client;

    private Bootstrap bootstrap;
    private ChannelFuture channelFuture;
    /** 用户下线监听服务 */
    private Bootstrap ntfLogoutBootstrap;
    private ChannelFuture ntfLogoutChannelFuture;
    private PortalClient ntfLogoutServer;
    // 监听AC发送的NTF_LOGOUT消息 端口
    // 用户强制下线消息
    private int ntfLogoutPort = PortalContext.NTF_LOGOUT_PORT;
    /** 向AC UDP 2000发送portal消息的本地UDP端口  */
    private int port = -1;

    /** 认证类型，目前支持CHAP、PAP */
    private final String authType;
    /** 消息发送后，未收到响应的超时时间 */
    private final int msgTimeout;
    /** 每个用户认证流程的超时时间 */
    private final int authTimeout;
    /** key为用户IP地址，<br/>
     * AuthKey为用户一次认证过程中的唯一标识，
     * 包括portal server生成的serialNo 和 AC生成的reqID <br/>
     * 认证完成后，删除*/
    private final ConcurrentHashMap<UserIP, AuthInfo> userAuthKeys = new ConcurrentHashMap<>();
    /** 记录用户当前认证状态<br/>
     * 认证完成后，删除 */
    private final ConcurrentHashMap<UserIP, AuthStatus> userStatus = new ConcurrentHashMap<>();
    /** 当前在线用户 */
    private final List<UserIP> onlines = new ArrayList<UserIP>();

    /** AC portal协议监听地址 */
    private static InetSocketAddress acAddress;
    /** 超时定时任务服务 */
    private ScheduledExecutorService timeoutScheduler =
            ThreadUtils.newDaemonSingleThreadScheduledExecutor("portal-server-timeout");
    /** 请求认证的处理线程池 */
    private ThreadPoolExecutor authExecutor =
            ThreadUtils.newDaemonCachedThreadPool("user-auth-async");

    public PortalServer() {
        conf =  new TransportConf("portal", new ConfigProvider() {
            @Override
            public String get(String name) {
                return sparkConf.get(name);
            }
        });
        context = new PortalContext(conf);

        // 读取用户配置的接受用户异常下线消息
        // 监听端口
        ntfLogoutPort = context.ntfLogoutPort();
        // 获取认证类型，CHAP 或者 PAP
        authType = context.authType();
        // 获取portal 消息响应超时时间
        msgTimeout = context.msgTimeoutInMill();
        authTimeout = context.authTimeoutInMill();
        acAddress = new InetSocketAddress("192.168.3.100", 2000);
        try {
            init();
        } catch (RuntimeException e) {
            JavaUtils.closeQuietly(this);
            throw e;
        }
    }

    public PortalServer(PortalContext context,
                        String hostToBind,
                        int portToBind) {
        this.context = context;
        this.conf = context.getConf();

        // 读取用户配置的接受用户异常下线消息
        // 监听端口
        ntfLogoutPort = context.ntfLogoutPort();
        // 获取认证类型，CHAP 或者 PAP
        authType = context.authType();
        // 获取portal 消息响应超时时间
        msgTimeout = context.msgTimeoutInMill();
        authTimeout = context.authTimeoutInMill();
        acAddress = new InetSocketAddress("192.168.3.100", 2000);
        try {
            init();
        } catch (RuntimeException e) {
            JavaUtils.closeQuietly(this);
            throw e;
        }
    }

    public int getPort() {
        if (port == -1) {
            throw new IllegalStateException("Server not initialized");
        }
        return port;
    }

    private void init() {
        initPortalClient();
        initNtfLogoutServer();
    }

    private void initNtfLogoutServer() {
        IOMode ioMode = IOMode.valueOf(conf.ioMode());
        EventLoopGroup bossGroup =
                NettyUtils.createEventLoop(ioMode, conf.serverThreads(), "notification-logout-server");

        PooledByteBufAllocator allocator = NettyUtils.createPooledByteBufAllocator(
                conf.preferDirectBufs(), true /* allowCache */, conf.serverThreads());

        ntfLogoutBootstrap = new Bootstrap()
                .group(bossGroup)
                .channel(NettyUtils.getDatagramChannelClass(ioMode))
                .option(ChannelOption.ALLOCATOR, allocator);

        if (conf.backLog() > 0) {
            ntfLogoutBootstrap.option(ChannelOption.SO_BACKLOG, conf.backLog());
        }

        if (conf.receiveBuf() > 0) {
            ntfLogoutBootstrap.option(ChannelOption.SO_RCVBUF, conf.receiveBuf());
        }

        if (conf.sendBuf() > 0) {
            ntfLogoutBootstrap.option(ChannelOption.SO_SNDBUF, conf.sendBuf());
        }

        ntfLogoutBootstrap.handler(new ChannelInitializer<DatagramChannel>() {
            @Override
            protected void initChannel(DatagramChannel ch) throws Exception {
                context.initializePipeline(ch, PortalServer.this);
                ntfLogoutServer = new PortalClient(ch);
            }
        });

        ntfLogoutChannelFuture = startServiceOnPort(ntfLogoutPort, ntfLogoutBootstrap);

        ntfLogoutPort = ((InetSocketAddress) channelFuture.channel().localAddress()).getPort();
        logger.debug("monitor notification logout message on port: " + ntfLogoutPort);
    }

    private void initPortalClient() {
        IOMode ioMode = IOMode.valueOf(conf.ioMode());
        EventLoopGroup bossGroup =
                NettyUtils.createEventLoop(ioMode, conf.serverThreads(), "portal-client");

        PooledByteBufAllocator allocator = NettyUtils.createPooledByteBufAllocator(
                conf.preferDirectBufs(), true /* allowCache */, conf.serverThreads());

        bootstrap = new Bootstrap()
                .group(bossGroup)
                .channel(NettyUtils.getDatagramChannelClass(ioMode))
                .option(ChannelOption.ALLOCATOR, allocator);

        if (conf.backLog() > 0) {
            bootstrap.option(ChannelOption.SO_BACKLOG, conf.backLog());
        }

        if (conf.receiveBuf() > 0) {
            bootstrap.option(ChannelOption.SO_RCVBUF, conf.receiveBuf());
        }

        if (conf.sendBuf() > 0) {
            bootstrap.option(ChannelOption.SO_SNDBUF, conf.sendBuf());
        }

        bootstrap.handler(new ChannelInitializer<DatagramChannel>() {
            @Override
            protected void initChannel(DatagramChannel ch) throws Exception {
                context.initializePipeline(ch, PortalServer.this);
                client = new PortalClient(ch);
            }
        });

//        channelFuture = bootstrap.connect(acAddress);
        channelFuture = startServiceOnPort(0, bootstrap);

        port = ((InetSocketAddress) channelFuture.channel().localAddress()).getPort();
        logger.debug("connected to AC on port: " + port);
    }

    @Override
    public void close() throws IOException {
        if (channelFuture != null) {
            // close is a local operation and should finish within milliseconds; timeout just to be safe
            channelFuture.channel().close().awaitUninterruptibly(10, TimeUnit.SECONDS);
            channelFuture = null;
        }
        if (bootstrap != null && bootstrap.group() != null) {
            bootstrap.group().shutdownGracefully();
        }
        bootstrap = null;

        client.close();
        client = null;

        // 关闭用户下线通知监听端口
        if (ntfLogoutChannelFuture != null) {
            ntfLogoutChannelFuture.channel().close().awaitUninterruptibly(10, TimeUnit.SECONDS);
            ntfLogoutChannelFuture = null;
        }
        if (ntfLogoutBootstrap != null && ntfLogoutBootstrap.group() != null) {
            ntfLogoutBootstrap.group().shutdownGracefully();
        }
        ntfLogoutBootstrap = null;

        ntfLogoutServer.close();
        ntfLogoutServer = null;
    }

    public static InetSocketAddress getAcAddress() {
        return acAddress;
    }

    @Override
    public void handle(PortalMessage message) throws Exception {
        if (message.type() == Type.LogoutNTF) {
            processNtfLogout(message);
            return;
        }
        Type msgType = message.type();

        AuthStatus receive = AuthStatus.getStatusByPortalMessage(message);
        AuthStatus current = userStatus.get(message.userIP());

        if (current == null) {
            // 如果该用户认证流程已被终止，
            logger.warn("{}'s auth has been terminated!", message.userIP());
            // 如果在发出AuthRequest之后用户流程被取消，
            // 收到认证成功响应报文时，还是应该更新用户在线列表
            if (isAuthAccept(message)) {
                synchronized (this) {
                    onlines.add(message.userIP());
                }
            }
            return;
        }
        if (receive.getStatus() <= current.getStatus()) {
            // receive报文是超时重发的报文，已经接受处理过，则丢弃
            logger.warn("{}'s current status is {}, but received status is !",
                    message.userIP(), current, receive);
            return;
        } else {
            // 更新 用户认证状态
            userStatus.put(message.userIP(), receive);
        }

        AuthInfo key = userAuthKeys.get(message.userIP());
        if(key == null || key.serialNo != message.serialNumber()) {
            logger.warn("{}'s serialNo is {}, but received message's serialNo is {}"
                    , message.userIP(), key.serialNo, message.serialNumber());
            return;
        }

        switch (msgType) {
            case ChallengeAck:
                processChallengeAck(message);
                break;
            case AuthAck:
                processAuthAck(message);
                break;
            case LogoutAck:
                processLogoutAck(message);
                break;
            default:
                logger.warn("error portal message type from ac! type = " + msgType.name());
        }

    }

    /**
     * 是认证通过报文
     * @param message
     * @return
     */
    private boolean isAuthAccept(PortalMessage message) {
        if (message.type() == Type.AuthAck &&
                message.errorCode() == ErrorCode.AuthSuccess) {
            return true;
        } else {
            return false;
        }
    }

    /**
     * 处理用户下线通知信息
     * @param message
     */
    private void processNtfLogout(PortalMessage message) {
        logger.info("{} has logout!", message.userIP());
        // 删除用户在线
        synchronized (this) {
            onlines.remove(message.userIP());
        }
    }

    /**
     * 用户下线请求响应 消息处理
     * @param message
     */
    private void processLogoutAck(PortalMessage message) {
        UserIP userIP = message.userIP();
        ErrorCode result = message.errorCode();

        AuthInfo info = userAuthKeys.get(userIP);
        switch (result) {
            case LogoutSuccess:
                logger.info("{} logout successful in {} mill-seconds"
                        , userIP, System.currentTimeMillis() - info.startTime);
                // 删除用户在线
                info.isSuccessful = true;
                synchronized (this) {
                    onlines.remove(userIP);
                }
                break;
            case LogoutReject:
                logger.warn("{}'s logout request has been reject", userIP);
                info.isSuccessful = false;
                break;
            case LogoutFailure:
                logger.warn("{}'s logout request has failure", userIP);
                info.isSuccessful = false;
                break;
            default:
                logger.warn("unsupported result code {} of " +
                        "logout ack message for {}", result, userIP);
                info.isSuccessful = false;
        }

        removeUserAuthAndSetResult(message.userIP());
    }

    private void processAuthAck(PortalMessage message) {
        UserIP userIP = message.userIP();
        ErrorCode resultCode = message.errorCode();

        AuthInfo info = userAuthKeys.get(userIP);
        switch (resultCode) {
            case AuthSuccess:
                // 认证成功
                logger.info("{} has been auth successful in {} mill-seconds!",
                        userIP, System.currentTimeMillis() - info.startTime);
                info.isSuccessful = true;
//                removeUserAuthAndSetResult(userIP);
                // 添加到在线用户组中去
                synchronized (this) {
                    onlines.add(userIP);
                }
                break;
            case AuthExist:
                // 用户已认证
                logger.warn("{} has been auth already!");
                // 判断用户是否已存在于在线列表中
                synchronized (this) {
                    if (!onlines.contains(userIP)) {
                        onlines.add(userIP);
                    }
                }
                info.isSuccessful = true;
//                removeUserAuthAndSetResult(userIP);
                break;
            case AuthReject:
            case AuthBusy:
            case AuthFailure:
                // 拒绝、忙、错误，
                // 都按照 认证失败处理
                info.isSuccessful = false;
//                removeUserAuthAndSetResult(userIP);
                logger.warn("{} auth failure, the result code is {} !", userIP, resultCode);

                break;
            default:
                logger.error("unsupported result code in AuthAck message: {}", resultCode);
                logger.trace(message.toString());
                info.isSuccessful = false;
//                removeUserAuthAndSetResult(userIP);
        }
        // 向AC发送 认证响应确认报文
        PortalMessage akk = PortalMessageFactory.createAuthAckAffMessage(
                message.authType(), userIP, info.serialNo, info.reqID);
        akk.setAcAddress(info.acAddress);
        client.send(akk, this);
        removeUserAuthAndSetResult(userIP);
    }

    /**
     *  删除用户认证信息,
     *  并设置认证返回结果给用户
     * @param userIP
     */
    private void removeUserAuthAndSetResult(UserIP userIP) {
        AuthInfo info = userAuthKeys.get(userIP);

        userAuthKeys.remove(userIP);
        userStatus.remove(userIP);

        // 将认证结果返回给用户
        if (info.result != null) {
            DeferredResult<ModelAndView> result = info.result;
            ModelAndView mav = new ModelAndView("portal-result");

            if (info.isAuth) {
                // 认证请求结果
                if (info.isSuccessful) {
                    // 认证成功
                    mav.addObject("title", "认证成功");
                    mav.addObject("wlanacip", info.acAddress.getHostName());
                    mav.addObject("result", "认证成功");
                } else {
                    // 认证失败
                    mav.addObject("title", "认证失败");
                    mav.addObject("result", "认证失败");
                }
            } else {
                // 下线请求结果
                if (info.isSuccessful) {
                    mav.setViewName("portal-login");
                    // 下线成功
                    mav.addObject("wlanacip", info.acAddress.getHostName());
                    mav.addObject("title", "下线成功");
                    mav.addObject("result", "下线成功");
                } else {
                    // 下线失败
                    mav.addObject("title", "下线失败");
                    mav.addObject("result", "下线失败");
                }

            }
            result.setResult(mav);
        }
    }

    private void processChallengeAck(PortalMessage message) {
        UserIP userIP = message.userIP();
        AuthInfo info = userAuthKeys.get(userIP);
        info.reqID = message.requestID();

        ErrorCode resultCode = message.errorCode();
        switch (resultCode) {
            case ChallengeSuccess:
                // 获取Challenge成功
                logger.info("{} get challenge successful in {} mill-seconds!",
                        userIP, System.currentTimeMillis() - info.startTime);
                PortalAttribute challengeAttr = message.attributes().get(0);
                if (challengeAttr.getType() != PortalAttribute.Type.Challenge) {
                    logger.warn("the challenge ack message's attribute type isn't " +
                            "challenge attribute! attribute type = " + challengeAttr.getType());
                    logger.trace(message.toString());

                    removeUserAuthAndSetResult(message.userIP());
                    return;
                }

                byte chapID = (byte) (message.requestID() & 0xff);
                byte[] challenge = challengeAttr.getValue();
                byte[] password = info.password.getBytes(StandardCharsets.UTF_8);
                try {
                    byte[] chapPassword = CHAP.chapMD5(chapID, password, challenge);
                    // 发送认证请求报文
                    PortalMessage msg = PortalMessageFactory.createCHAPAuthRequestMessage(message.userIP(),
                            message.serialNumber(), message.requestID(), info.userName, chapPassword);
                    msg.setAcAddress(info.acAddress);
                    client.send(msg, this);
                } catch (NoSuchAlgorithmException e) {
                    logger.warn("doesn't support md5 algorithm!", e);
                    removeUserAuthAndSetResult(message.userIP());
                    return;
                }
                break;
            case ChallengeExist:Exist:
                // 用户Challenge已存在
                logger.warn("{} has got challenge already!", userIP);
                removeUserAuthAndSetResult(userIP);
                break;
            case ChallengeReject:
            case ChallengeBusy:
            case ChallengeFailure:Failure:
                // 拒绝、忙、错误，
                // 都按照 认证失败处理
                removeUserAuthAndSetResult(userIP);
                logger.warn("{} get challenge failure, the result code is {} !", userIP, resultCode);

                break;
            default:
                logger.error("unsupported result code in ChallengeAck message: {}", resultCode);
                logger.trace(message.toString());
                removeUserAuthAndSetResult(userIP);
        }
    }

    @Override
    public void channelActive() {
    }

    @Override
    public void exceptionCaught(Throwable cause) {

    }

    /**
     * 和AC的通道关闭后，需要重新建立一条通道 <br/>
     * 监听用户下线通知的通道关闭后，需要重新建立一条
     */
    @Override
    public void channelInactive() {
        // 判断是否portal client通道挂了
        if (channelFuture != null && !channelFuture.channel().isActive()) {
            client.close();
            bootstrap.group().shutdownGracefully();
            initPortalClient();
        }
        // 判断是否用户下线监听通道挂了
        if (ntfLogoutChannelFuture != null &&
                !ntfLogoutChannelFuture.channel().isActive()) {
            ntfLogoutServer.close();
            ntfLogoutBootstrap.group().shutdownGracefully();
            initNtfLogoutServer();
        }
    }

    /**
     * 发送potal消息失败后，需要清理该用户的认证信息，
     * 并向用户返回认证失败及原因
     * @param userIP
     * @param msgType
     * @param e
     */
    @Override
    public void onFailure(PortalMessage.UserIP userIP, PortalMessage.Type msgType, Throwable e) {
        logger.warn("send {}'s {} message fail!", userIP, msgType);
        logger.warn("error cause:", e);

        // 删除该用户的认证信息
        removeUserAuthAndSetResult(userIP);
    }


    /**
     * 报文发送成功后，需要启动接受相应报文超时机制
     * @param msg
     */
    public void onSuccess(final PortalMessage msg) {
        // 向AC发送的超时消息、认证响应确认消息
        // 不进行更新处理
        if (isTimeoutMessage(msg) ||
                msg.type() == Type.AuthAckAff){
            return;
        }

        logger.info("send portal message {} successful " +
                "for {}", msg.type(), msg.userIP());

        // 更新用户认证状态
        if (userStatus.get(msg.userIP()) == null) {
            // status为空，说明该用户的认证流程已被终止
            logger.warn("{}'s status is null", msg.userIP());
            return;
        }
        final AuthStatus status = AuthStatus.getStatusByPortalMessage(msg);
        userStatus.put(msg.userIP(), status);

        PortalMessage timeoutMessage = null;
        if (msg.type() == Type.AuthReq) {
            // 认证请求超时消息的serialNo, reqID
            // 和 请求消息保持一致
            timeoutMessage = PortalMessageFactory
                    .createTimeoutMessage(msg.authType(), msg.serialNumber()
                            , msg.requestID(), msg.userIP());
        } else if (msg.type() == Type.ChallengeReq ||
                msg.type() == Type.LogoutReq) {
            // 保持serialNo, reqID写0
            timeoutMessage = PortalMessageFactory
                    .createTimeoutMessage(msg.authType(), msg.serialNumber(), 0, msg.userIP());
        }

        final PortalMessage message = timeoutMessage;
        if (timeoutMessage != null) {
            timeoutScheduler.schedule(new Runnable() {
                @Override
                public void run() {
                    AuthStatus currentStatus = userStatus.get(msg.userIP());
                    if (currentStatus != null && currentStatus == status) {
                        logger.warn("have not received the response message " +
                                "in {} mill-seconds for {} message about {}", msgTimeout, msg.type(), msg.userIP());
                        message.setAcAddress(msg.getAcAddress());
                        client.send(message, PortalServer.this);
                    }
                }
            }, msgTimeout, TimeUnit.MILLISECONDS);
        }
    }

    private boolean isTimeoutMessage(PortalMessage msg) {
        if (msg.type() == Type.LogoutReq && msg.errorCode() == ErrorCode.Timeout) {
            return true;
        }

        return false;
    }

    /**
     * 开始用户下线
     * @param userIP
     */
    public void startUserLogout(String acip, String userIP) {
//        // 判断用户是否在线
//        // 不在线，则不处理
//        if (!isUserOnline(userIP)) {
//            logger.warn("{} isn't online now," +
//                    " can't start logout process!", userIP);
//            return;
//        }

        // 写入用户下线请求状态
        final UserIP user = new UserIP(userIP);
        userStatus.put(user, AuthStatus.LogoutReq);

        PortalMessage send = null;
        if (authType.equalsIgnoreCase(PortalContext.AUTH_TYPE_CHAP)) {
            // CHAP认证
            send = PortalMessageFactory.createLogoutRequestMessage(AuthType.CHAP, user);
        } else {
            // PAP认证
            send = PortalMessageFactory.createLogoutRequestMessage(AuthType.PAP, user);
        }

        // 写入用户认证Key
        AuthInfo key = new AuthInfo(acip, null, null, send.serialNumber());
        userAuthKeys.put(user, key);
        // 标识 下线请求
        key.isAuth = false;

        send.setAcAddress(key.acAddress);

        client.send(send, this);
        // 开启认证超时定时任务
        timeoutScheduler.schedule(new Runnable() {
            @Override
            public void run() {
                AuthStatus currentStatus = userStatus.get(user);
                if (currentStatus != null && currentStatus.getStatus()
                        < AuthStatus.LogoutAck.getStatus()) {
                    logger.warn("have not get the logout result " +
                            "in {} mill-seconds for {}, current status is {}"
                            , authTimeout,  user, currentStatus);
                    removeUserAuthAndSetResult(user);
                }
            }
        }, authTimeout, TimeUnit.MILLISECONDS);

        logger.info("starting logout for {}", userIP);
    }

//    /**
//     * 用户是否已在线
//     * @param userIP
//     * @return
//     */
//    private boolean isUserOnline(String userIP) {
//        return onlines.contains(userIP);
//    }

    /**
     * 开始用户认证
     * @param acip
     * @param userIP
     * @param userName
     * @param password
     */
    public void startUserAuth(String acip, String userIP, String userName, String password) {
        // 判断用户IP是否正在认证，
        // 只基于IP判断，同一用户在不同IP地址上,
        // 判断为不同的认证流程，可以开启认证
        if (isUserInAuthing(userIP)) {
            logger.warn("{} is already in auth process " +
                    "by user {}", userIP, userName);
            return;
        }
//        // 判断用户是否已在线
//        // 在线则不开启认证流程
//        if (isUserOnline(userIP)) {
//            logger.warn("{} is already online, can't auth again!", userIP);
//            return
//        }

        // 写入用户认证开始状态
        final UserIP user = new UserIP(userIP);
        userStatus.put(user, AuthStatus.Started);

        PortalMessage send = null;
        if (authType.equalsIgnoreCase(PortalContext.AUTH_TYPE_CHAP)) {
            // CHAP认证
            send = PortalMessageFactory.createChallengeMessage(userIP);
        } else {
            // PAP认证
            send = PortalMessageFactory.createPAPAuthRequestMessage(user, userName, password);
        }

        // 写入用户认证Key
        AuthInfo key = new AuthInfo(acip, userName, password, send.serialNumber());
        userAuthKeys.put(user, key);

        send.setAcAddress(key.acAddress);

        client.send(send, this);
        // 开启认证超时定时任务
        timeoutScheduler.schedule(new Runnable() {
            @Override
            public void run() {
                AuthStatus currentStatus = userStatus.get(user);
                if (currentStatus != null && currentStatus.getStatus()
                        < AuthStatus.AuthAck.getStatus()) {
                    logger.warn("have not get the auth result " +
                            "in {} mill-seconds for {}, current status is {}"
                            , authTimeout,  user, currentStatus);
                    removeUserAuthAndSetResult(user);
                }
            }
        }, authTimeout, TimeUnit.MILLISECONDS);

        logger.info("starting auth for {}", userIP);
    }

    /**
     * 判断用户IP是否正在认证流程中
     * @param userIP
     * @return
     */
    private boolean isUserInAuthing(String userIP) {
        AuthInfo info = userAuthKeys.get(new UserIP(userIP));
        return info != null;
    }

    /**
     * Attempt to start a service on the given port, or fail after a number of attempts.
     * Each subsequent attempt uses 1 + the port used in the previous attempt (unless the port is 0).
     *
     * @param startPort The initial port to start the service on.
     */
    private ChannelFuture startServiceOnPort(
    int startPort,
    Bootstrap bootstrap) {
        // "startPort should be between 1024 and 65535 (inclusive), or 0 for a random free port."
        assert (startPort == 0 || (1024 <= startPort && startPort < 65536));

        int maxRetries = context.portMaxRetries();
        for (int offset = 0; offset < maxRetries; offset++) {
            // Do not increment port if startPort is 0, which is treated as a special port
            int tryPort;
            if (startPort == 0) {
                tryPort = 0;
            } else {
                // If the new port wraps around, do not try a privilege port
                tryPort = ((startPort + offset - 1024) % (65536 - 1024)) + 1024;
            }
            try {
                ChannelFuture future = bootstrap.bind(tryPort);
                future.syncUninterruptibly();
                logger.info("Successfully started service on port {}.", tryPort);
                return future;
            } catch(Exception e) {
                if (isBindCollision(e)) {
                    if (offset >= maxRetries) {
                        String exceptionMessage = e.getMessage() + ": Service failed after " +
                                maxRetries + " retries!";
                        BindException exception = new BindException(exceptionMessage);
                        // restore original stack trace
                        exception.setStackTrace(e.getStackTrace());
                        return bootstrap.bind(0);
                    }
                }
                logger.warn("Service could not bind on port {}. " +
                        "Attempting port {}.", tryPort, tryPort + 1);
            }
        }
        // Should never happen
        logger.error("Failed to start service$serviceString on port {}", startPort);
        return null;
    }

    /**
     * Return whether the exception is caused by an address-port collision when binding.
     */
    private boolean isBindCollision(Throwable exception) {
        if (exception instanceof BindException) {
            if (exception.getMessage() != null) {
                return true;
            }

            return isBindCollision(exception.getCause());
        } else if (exception instanceof MultiException) {
            MultiException me = (MultiException) exception;
            for (Throwable e : me.getThrowables()) {
                if (isBindCollision((e))) {
                    return true;
                }
            }
            return false;
        } else if (exception instanceof Exception) {
            return isBindCollision(exception.getCause());
        }

        return false;
    }

    public void startUserAuthAsyncCall(final DeferredResult<ModelAndView> deferredResult
            , final String acip, final String ip, final String username, final String password) {
        authExecutor.submit(new Runnable() {
            @Override
            public void run() {
                startUserAuth(acip, ip, username, password);

                AuthInfo info = userAuthKeys.get(new UserIP(ip));
                info.result = deferredResult;
            }
        });
    }

    public void startUserLogoutAsyncCall(final DeferredResult<ModelAndView> result, final String acip, final String userIP) {
        authExecutor.submit(new Runnable() {
            @Override
            public void run() {
                startUserLogout(acip, userIP);

                AuthInfo info = userAuthKeys.get(new UserIP(userIP));
                info.result = result;
            }
        });
    }
}
